//
// Created by benxb on 2021/11/27.
//

#include "storage/data_chunk.h"
#include "common/logger.h"
#include <iostream>

namespace benxdb {

bool DataChunk::AppendRecord(const Value& value) {
    if (value.GetTypeId() != data_type_) {
        LOG_ERROR("Wrong data_ type.");
        return false;
    }
    if (chunk_size_ + value.GetLength() > capacity_) {
        LOG_ERROR("Chunk size out of bound.");
        return false;
    }

    memcpy(data_+next_slot_, value.GetData(), value.GetLength());
    chunk_size_ += value.GetLength();
    next_slot_ += value.GetLength();
    record_count_++;
    return true;
}

bool DataChunk::AppendManyRecords(const std::vector<Value>& records) {
    if (records.empty()) {
        LOG_ERROR("Empty records.");
        return false;
    }
    if (records[0].GetTypeId() != data_type_) {
        LOG_ERROR("Wrong data_ type.");
        return false;
    }
    if (chunk_size_ + records[0].GetLength() * records.size() > CHUNK_SIZE) {
        LOG_ERROR("Chunk size out of bound.");
        return false;
    }

    for (const auto& value : records) {
        memcpy(data_ + next_slot_, value.GetData(), value.GetLength());
        chunk_size_ += value.GetLength();
        next_slot_ += value.GetLength();
    }
    return true;
}

void DataChunk::GetMaxMinValue(Value &max_val, Value &min_val) {
    switch (data_type_) {
        case INVALID:
            break;
        case BOOLEAN:
            GetMMValue<int8_t>(max_val, min_val);
            break;
        case TINYINT:
            GetMMValue<int8_t>(max_val, min_val);
            break;
        case SMALLINT:
            GetMMValue<int16_t>(max_val, min_val);
            break;
        case INTEGER:
            GetMMValue<int32_t>(max_val, min_val);
            break;
        case BIGINT:
            GetMMValue<int64_t>(max_val, min_val);
            break;
        case FLOAT:
            GetMMValue<float>(max_val, min_val);
            break;
        case DECIMAL:
            GetMMValue<double>(max_val, min_val);
            break;
        case VARCHAR:
            break;
        case TIMESTAMP:
            GetMMValue<uint64_t>(max_val, min_val);
            break;
    }
}

std::vector<Value> DataChunk::QueryRange(const Value& min_val, const Value& max_val) {
    std::vector<Value> res;
    int value_len = GetValueSize();
    for (int offset = 0; offset <= chunk_size_; offset += value_len) {
        Value val(data_type_, data_+offset);
        if (val.CompareLessThanEquals(max_val) == CmpBool::CmpTrue &&
            val.CompareGreaterThanEquals(min_val) == CmpBool::CmpTrue) {
            res.push_back(val);
        }
    }
    return res;
}

std::vector<std::pair<int32_t, Value>> DataChunk::QueryIndexValueRange(Value min_val, Value max_val) {
    std::vector<std::pair<int32_t, Value>> res;
    int value_len = GetValueSize();
    for (int index = 0; index <= chunk_size_; index ++) {
        Value val(data_type_, data_+index*value_len);
        if (val.CompareLessThanEquals(max_val) == CmpBool::CmpTrue &&
            val.CompareGreaterThanEquals(min_val) == CmpBool::CmpTrue) {
            res.emplace_back(index, val);
        }
    }
    return res;
}

Value DataChunk::QueryIndex(int32_t index) {
    int val_len = GetValueSize();
    if (index > chunk_size_/val_len) {
        LOG_ERROR("index out of bound");
        return Value(TypeId::INVALID);
    }
    return Value(data_type_, data_ + index*val_len);
}

void DataChunk::Clear() {
    next_slot_ = 0;
    chunk_size_ = 0;
}

//bool DataChunk::AppendData(const char *data, int size) {
//    if (chunk_size_ + size > capacity_) {
//        LOG_ERROR("Chunk size out of bound.");
//        return false;
//    }
//
//    memcpy(data_+next_slot_, data, size);
//    chunk_size_ += size;
//    next_slot_ += size;
//    record_count_++;
//    return true;
//}

template<typename T>
void DataChunk::GetMMValue(Value &max_val, Value &min_val) {
    int32_t value_len = GetValueSize();
    T max_t, min_t;
    max_t = min_t = *reinterpret_cast<T*>(data_);
    for (int offset = 0; offset < chunk_size_; offset += value_len) {
        T val = *reinterpret_cast<T*>(data_+offset);
        max_t = std::max(max_t, val);
        min_t = std::min(min_t, val);
    }
    max_val = Value(data_type_, max_t);
    min_val = Value(data_type_, min_t);
}
}